package org.deng.learn.lambda;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author 邓力宾
 * @date 2020/12/18 11:52
 * @desc 自定义处理器
 */
public class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<Integer, String> {

    private Flow.Subscription subscription;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        //请求一个数据
        this.subscription.request(1);
    }

    /**
     * 接收到一个数据
     * @param item
     */
    @Override
    public void onNext(Integer item) {
        System.out.println("处理器处理数据:"+item);
        //过滤到小于0的数据，再发布出去
        if(item > 0){
            this.submit("转换后的数据:"+item);
        }
        //请求下一个
        this.subscription.request(1);

        //或者已经达到目标了，调用cancel告诉发布者不再接收数据了
        //this.subscription.cancel();
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
        System.out.println("处理器处理数据异常");
        this.subscription.cancel();
    }

    @Override
    public void onComplete() {
        System.out.println("处理器已完成");
    }
}
